/*
 * sccsim.cpp, SmartCam coprocessor library for simulation
 *
 */

#define _USE_MATH_DEFINES

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <math.h>
#include <getopt.h>

#include <vector>
#include <fstream>

#include "CStr.h"
#include <memfs.h>

#include <des.hpp>
#include <devices.hpp>
#include <log.hpp>

#include <scos.h>

#include "sc.h"
#include "settings.h"
#include "scc.h"
#include "scu.h"
#include "scsim.h"

class simThread : public DESthread
{
  public:
    simThread(std::string name, void *arg) : DESthread(name, arg) { }

    virtual void main(void *arg);

  public:
    class simProcessor *m_sdevice;
    sc_proc_t *m_processor;
};

class simMainThread : public simThread
{
  public:
    simMainThread(std::string name, void *arg) : simThread(name, arg) { }

    void main(void *arg);
};

class simProcessor : public virtual DESdevice
{
  public:
    virtual void addProcess(simThread* process) = 0;

  public:
    sc_proc_t *m_processor;
    std::list<scc_msg_t> m_queue;
    std::list<scc_datamsg_t> m_dataqueue;
    DESsignal m_signal;
};

class simSequentialProcessor : public simProcessor, public DESroundRobinDevice
{
  public:
    void processEvent(DESevent event)
    {
      DESroundRobinDevice::processEvent(event);
    }

    DESevent wait(class DESprocess *process, class DESsignal* signal)
    {
      return DESroundRobinDevice::wait(process, signal);
    }

    DESevent wait(class DESprocess *process, TIME time)
    {
      return DESroundRobinDevice::wait(process, time);
    }

    void exit(class DESprocess *process)
    {
      DESroundRobinDevice::exit(process);
    }

    void addProcess(simThread* process)
    {
      DESroundRobinDevice::addProcess(process);

      process->m_sdevice = this;
      process->m_processor = m_processor;
    }
};

class simParallelProcessor : public simProcessor, public DESparallelDevice
{
  public:
    void processEvent(DESevent event)
    {
      DESparallelDevice::processEvent(event);
    }

    DESevent wait(class DESprocess *process, class DESsignal* signal)
    {
      return DESparallelDevice::wait(process, signal);
    }

    DESevent wait(class DESprocess *process, TIME time)
    {
      return DESparallelDevice::wait(process, time);
    }

    void exit(class DESprocess *process)
    {
      DESparallelDevice::exit(process);
    }

    void addProcess(simThread* process)
    {
      DESparallelDevice::addProcess(process);
      
      process->m_sdevice = this;
      process->m_processor = m_processor;
    }
};

extern "C" void simReceiveMessage(scc_hdr_t *hdr);
extern "C" void simReceiveDataMessage(scc_hdr_t *hdr);
void simProcessMessage(scc_hdr_t *hdr);
void *simOperationEnqueue(void *op);
sim_op_t *simOperationAllocate(void);
sim_op_t *simOperationFind(int id);
void simOperationFree(sim_op_t *op);
void simWriteResult(int p, sim_op_t *o, int id, int type, char *data);
void simOperationComplete(int p, sim_op_t *o);
extern "C" void simBufferDestructor(scu_buf_t *buf);
extern "C" void simTransportReceiver(scc_hdr_t *hdr);
extern "C" unsigned long int simBufferQuery(int buffer);
extern "C" void simTransportTransmitter(int src, int dest, int destbuf, int bytes);
void simCustomOperationEnqueue(int proc, sim_op_t *op);
void simOperation(int proc, sim_op_t *op, int arguments);

static sim_op_t  sim_op[SC_MAX_OPERATIONS];
static scu_buf_t sim_buf[SC_MAX_BUFFERS];
static int       sim_weight[SC_MAX_BUFFERS];

static scos_mutex_t sim_mutex, sim_buf_mutex;
static scos_cond_t  sim_cond, sim_buf_cond;

CLog siml("siml");
static std::vector<simProcessor*> sim_procs;

static Mfs trace_fs;
static Mshell trace(trace_fs);

#ifdef DES
extern DESroundRobinDevice MCP;
#endif

off_t fsize(const char *file_name)
{
  struct stat buf;

  abortOnError(STREAM(siml.err.GetStream()), stat(file_name, &buf));

  return buf.st_size;
}

extern "C" void scSimInit(sc_proc_t *p)
{
  static int first = 1;
  int i;
  simProcessor *sp;
  scs_target_t *t;

  if (first)
  {
    siml.debug.Edit(LOG_GROUP_SET_LEVEL);
    for (i = 0; i < SC_MAX_BUFFERS; i++)
      sim_buf[i].references = 0;

    scos_mutex_init(&sim_mutex);
    scos_cond_init(&sim_cond);

    scos_mutex_init(&sim_buf_mutex);
    scos_cond_init(&sim_buf_cond);

    siml.notice() << HDR << "Reading simulation trace from " << settings.simulate << std::endl;
    
    std::ifstream inp(settings.simulate);
    XMLParser parser(inp);

    abortOnError(STREAM(siml.err.GetStream()), !inp.is_open());
    abortOnError(STREAM(siml.err.GetStream()), trace.Import(parser, "/"));

    first = 0;
  }

  siml.notice() << HDR << "Initializing SmartCam SIM coprocessor " << p->id << std::endl;

  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&scs_targets, p->id, &t));

  if (t->parallel == 1)
    sp = new simParallelProcessor();
  else
    sp = new simSequentialProcessor();

  sp->m_processor = p;

  if (p->id >= (int)sim_procs.size())
    sim_procs.resize(p->id+1);

  sim_procs[p->id] = sp;
  sp->m_name = p->name;

  simMainThread *mt = new simMainThread("main", NULL);
  
  sp->addProcess(mt);

  sccTransportRegister(p->id, simTransportReceiver, simBufferQuery);
  sccCoprocessorSetup(p->id, simReceiveMessage, simReceiveDataMessage);
}

extern "C" void scSimExit(char *reportfile)
{
  FILE *f;
  double energy=0, time=DESgetCurrentTime()/1000000000.0;

  if (reportfile)
  {
    abortOnNULL(STREAM(siml.err.GetStream()), f = fopen(reportfile, "w"));
  }
  else
    f = stdout;

  fprintf(f, "<document>\n");
  fprintf(f, "  <E time=\"%f\"/>\n", time);

#ifdef DES
  fprintf(f, "  <%-20s><E load=\"%6.4f\"/></%-20s>\n",
	  "mcp",
	  MCP.getProcessedTime()*1.0/DESgetCurrentTime(),
	  "mcp");
#endif
  
  for (std::vector<simProcessor*>::iterator ii=sim_procs.begin();
       ii != sim_procs.end(); ++ii)
  {
    scs_target_t *t;

    if ((*ii) && strcmp(settings.name, (*ii)->m_name.c_str()))
    {
      /* Add idle power */
      abortOnNULL(STREAM(siml.err.GetStream()),
		  tblget(&scs_targets, (*ii)->m_processor->id, &t));
      if (t->power > 0)
	(*ii)->m_energy += t->power*time;

      simSequentialProcessor *sp = dynamic_cast<simSequentialProcessor*>((*ii));

      if (sp)
      {
	fprintf(f, "  <%-20s><E load=\"%6.4f\"/>\n                        ",
		(*ii)->m_name.c_str(), 
		sp->getProcessedTime()*1.0/DESgetCurrentTime());
      }
      else
	fprintf(f, "  <%-20s>", (*ii)->m_name.c_str());
      
      fprintf(f, "<E energy=\"%8.6f\"/></%-20s>\n",
	      (*ii)->m_energy,
	      (*ii)->m_name.c_str());
      energy += (*ii)->m_energy;
    }
  }

  fprintf(f, "  <E energy=\"%f\"/>\n", energy);
  fprintf(f, "  <E predicted-time=\"%f\"/>\n", last_makespan_prediction);
  fprintf(f, "  <E predicted-energy=\"%f\"/>\n", last_energy_prediction);

  fprintf(f, "</document>\n");
  if (reportfile) fclose(f);
}

extern "C" void simReceiveMessage(scc_hdr_t *hdr)
{
  simProcessor *p = sim_procs[hdr->destination];

  if (!strcmp(settings.name, p->m_processor->name))
  {
    /* *** Master processor. Handle in MCP context *** */
    simProcessMessage(hdr);
  }
  else
  {
    /* Other processor. Put in queue, and handle in the appropriate context */
    scc_msg_t msg;

    memcpy(&msg, hdr, sizeof(scc_msg_t));
    p->m_queue.push_back(msg);

    p->m_signal.signal();
  }
}

extern "C" void simReceiveDataMessage(scc_hdr_t *hdr)
{
  simProcessor *p = sim_procs[hdr->destination];

  if (!strcmp(settings.name, p->m_processor->name))
  {
    /* *** Master processor. Handle in MCP context *** */
    siml.err() << HDR << "Data message for MCP" << std::endl;
    sccTransportReceive(hdr);
  }
  else
  {
    /* Other processor. Put in queue, and handle in the appropriate context */
    if (hdr->type == SCC_MSGTYPE_BUFFERACK)
    {
      sccTransportReceive(hdr);
    }
    else
    {
      scc_datamsg_t datamsg;

      memcpy(&datamsg, hdr, sizeof(scc_datamsg_t));
      p->m_dataqueue.push_back(datamsg);

      p->m_signal.signal();
    }
  }
}

void simProcessMessage(scc_hdr_t *hdr)
{
  sim_op_t *op;
  scc_msg_t *msg = (scc_msg_t*)hdr;

  abortOnError(STREAM(siml.err.GetStream()), ((unsigned int)hdr->destination >= sim_procs.size()));

  simProcessor *p = sim_procs[hdr->destination], *p2;
  
  switch (msg->header.type)
  {
    case SCC_MSGTYPE_BUFFERCREATE:
      siml.debug() << HDR << "Creating buffer " << *((int*)msg->data) << " "
		   << &sim_buf[*((int*)msg->data)] << std::endl;
      abortOnBounds(STREAM(siml.err.GetStream()), *((int*)msg->data), 0, SC_MAX_BUFFERS);
      sim_buf[*((int*)msg->data)] = scuBufferCreate(msg->size,
					 	    SCU_BUF_MAX_READERS,
						    simBufferDestructor);
      scuBufferSetIndex(&sim_buf[*((int*)msg->data)], *((int*)msg->data));
      scos_mutex_lock(&sim_buf_mutex);
      scos_cond_signal(&sim_buf_cond);
      scos_mutex_unlock(&sim_buf_mutex);
      break;
    case SCC_MSGTYPE_BUFFERFINALIZE:
      siml.debug() << HDR << "Finalizing buffer " << *((int*)msg->data) << std::endl;;
      abortOnBounds(STREAM(siml.err.GetStream()), *((int*)msg->data), 0, SC_MAX_BUFFERS);
      scuBufferUnlock(&sim_buf[*((int*)msg->data)]);
      break;
    case SCC_MSGTYPE_TRANSPORTCREATE:
      p2 = sim_procs[msg->processor];

      siml.debug() << HDR << "Transmitting buffer " << *((int*)&msg->data[0])
		   << " from " << p->m_name << " to " << p2->m_name << "." << *((int*)&msg->data[sizeof(int)])
		   << " with weight " << msg->opid << std::endl;

      sim_weight[*((int*)&msg->data[sizeof(int)])] = msg->opid;

      sccTransportCreate(&sim_buf[*((int*)&msg->data[0])], hdr->destination, 
			 msg->processor, *((int*)&msg->data[sizeof(int)]));
      break;
    case SCC_MSGTYPE_OPERATIONENQUEUE:
      siml.debug() << HDR << "Enqueueing operation " << msg->opid << " type " << msg->optype 
		   << " (needs " << msg->size << " args)" << std::endl;
      abortOnBounds(STREAM(siml.err.GetStream()), msg->opid, 0, SC_MAX_OPERATIONS);
      op = &sim_op[msg->opid];
      op->id = msg->opid;
      op->operation = msg->optype;
      op->arguments = msg->size;      
      op->argreceived = 0;

      /* Arguments that are not sent are output arguments */
      for (int ii=0; ii < SC_MAX_ARGUMENTS; ii++)
      {
	op->argument[ii].id = ii;
	op->argument[ii].intent = SC_INTENT_OUT;
      }

      break;
    case SCC_MSGTYPE_OPERATIONARGUMENT:
      abortOnBounds(STREAM(siml.err.GetStream()), msg->opid, 0, SC_MAX_OPERATIONS);
      abortOnBounds(STREAM(siml.err.GetStream()), msg->argid, 0, SC_MAX_ARGUMENTS);
      
      op = &sim_op[msg->opid];
      op->argument[msg->argid].id = msg->argid;
      op->argument[msg->argid].type = msg->argtype;
      op->argument[msg->argid].intent = msg->argintent;

      switch (msg->argtype)
      {
	case SC_VAR_STREAM:
	  op->argument[msg->argid].data.d = *((int*)msg->data);
	  siml.debug() << HDR << "Received operation " << msg->opid << " STREAM argument "
		       << msg->argid << " with value " << op->argument[msg->argid].data.d
		       << std::endl;

          op->argument[msg->argid].buffer =
            &sim_buf[op->argument[msg->argid].data.d];

	  if (msg->argintent == SC_INTENT_IN)
	  {
	    /* To avoid adding a reader after the buffer has already been
	     * finalized (because the operation was waiting for another input),
	     * add the reader immediately.
	     */
	    
	    op->argument[msg->argid].reader =
	      scuBufferAddReader(op->argument[msg->argid].buffer);
	  }

	  break;
	case SC_VAR_INTEGER:
	  op->argument[msg->argid].data.d = *((int*)msg->data);
	  siml.debug() << HDR << "Received operation " << msg->opid << " INTEGER argument "
		       << msg->argid << " with value " << op->argument[msg->argid].data.d
		       << std::endl;
	  break;
	case SC_VAR_DOUBLE:
	  op->argument[msg->argid].data.f = *((double*)msg->data);
	  siml.debug() << HDR << "Received operation " << msg->opid << " DOUBLE argument "
		       << msg->argid << " with value " << op->argument[msg->argid].data.f
		       << std::endl;
	  break;
      }
      op->argreceived++;

      if (op->argreceived == op->arguments)
      {
	scs_operation_t *so;

	abortOnNULL(STREAM(siml.err.GetStream()),
		    tblget(&scs_operations, op->operation, &so));

	siml.info() << HDR << "Starting operation " << op->id << " on "
		    << p->m_processor->name << std::endl;

	scos_mutex_lock(&sim_mutex);

	simThread *t = new simThread((CStr(so->name) << "(" << op->id << ")").str(), (void*)op);

#ifdef DES
	if (!strcmp(settings.name, p->m_processor->name))
	{
	  /* Master processor. Attach thread to the MCP */
	  t->m_processor = p->m_processor;
	  MCP.addProcess(t);
	}
	else
#endif
	  p->addProcess(t);

	scos_cond_wait(&sim_cond, &sim_mutex);
	scos_mutex_unlock(&sim_mutex);

	siml.info() << HDR << "Operation started" << std::endl;
      }
      break;
   }
}

void simMainThread::main(void *dummy)
{
  logRegisterThread(CStr(m_processor->name) << "." << m_name);

  siml.info() << HDR << "Main thread started" << std::endl;

  while (1)
  {
    while (m_sdevice->m_queue.size())
    {
      simProcessMessage((scc_hdr_t*)&m_sdevice->m_queue.front());
      m_sdevice->m_queue.pop_front();
    }

    if (m_sdevice->m_dataqueue.size())
    {
      sccTransportReceive((scc_hdr_t*)&m_sdevice->m_dataqueue.front());
      m_sdevice->m_dataqueue.pop_front();
    }

    if (!m_sdevice->m_queue.size() && !m_sdevice->m_dataqueue.size())
      wait(&m_sdevice->m_signal);
  }
}

void simThread::main(void *operation)
{
  sim_op_t *op = (sim_op_t*)operation;  
  scs_target_t *t;
  scs_target_operation_t *to;
  
  scos_mutex_lock(&sim_mutex);
  scos_cond_signal(&sim_cond);
  scos_mutex_unlock(&sim_mutex);

  logRegisterThread(CStr(m_processor->name) << "." << m_name);

  siml.info() << HDR << "Started operation type " << op->operation << std::endl;

  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&scs_targets, m_processor->id, &t));
  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&t->operations, op->operation, &to));
  if (to->energy > 0) scos_dissipate(to->energy);

  simCustomOperationEnqueue(m_processor->id, op);

  siml.info() << HDR << "Finished" << std::endl;

  simOperationComplete(m_processor->id, op);
  simOperationFree(op);
}

void simCustomOperationEnqueue(int proc, sim_op_t *op)
{
  int opid=-1;
  int arguments;
  char *element, *element2;
  scs_operation_t *so;
  scs_target_operation_t *st;
  char buf[PATH_MAX] = {0};

  sprintf(buf, "Trying to find an operation matching %d (", op->operation);

  /* For all buffer inputs: read first input word; this contains the
   * stream identifier for that argument */
  for (int ii=0; ii != SC_MAX_ARGUMENTS; ++ii)
  {
    if (op->argument[ii].type == SC_VAR_STREAM)
    {
      if (op->argument[ii].intent == SC_INTENT_IN)
      {
	scuBufferRead(op->argument[ii].buffer, op->argument[ii].reader,
		      &op->argument[ii].stream, sizeof(op->argument[ii].stream),
		      SCU_BUF_BLOCK_ALL);

	sprintf(&buf[strlen(buf)], "i [%d], ", op->argument[ii].stream);
      }
      else
	sprintf(&buf[strlen(buf)], ", ");
    }
    else if (op->argument[ii].type == SC_VAR_INTEGER)
    {
      if (op->argument[ii].intent == SC_INTENT_IN)
	sprintf(&buf[strlen(buf)], "d [%d], ", op->argument[ii].data.d);
      else
	sprintf(&buf[strlen(buf)], ", ");
    }
    else if (op->argument[ii].type == SC_VAR_DOUBLE)
    {
      if (op->argument[ii].intent == SC_INTENT_IN)
	sprintf(&buf[strlen(buf)], "f [%f], ", op->argument[ii].data.f);
      else
	sprintf(&buf[strlen(buf)], ", ");
    }
  }

  buf[strlen(buf)-2] = ')';
  siml.debug() << HDR << buf << std::endl;

  /* Now find the first operation that has the same inputs and the same type */
  Mshell opshell(trace, "/trace/operations");

  opshell.FindInitDirectoryElements(".");
  while (!opshell.FindGetNextElement(&element) && opid == -1)
  {
    int type=-1, used=0;
    Mshell oshell(opshell, element);
    Mshell argshell(oshell, "arguments");

    arguments = 0;

    oshell.GetContent("used", &used);
    if (used) continue;

    oshell.GetContent("type", &type);
    if (type != op->operation) continue;

    oshell.GetContent("id", &opid);

    argshell.FindInitDirectoryElements(".");
    while (!argshell.FindGetNextElement(&element2))
    {
      int id=-1, type=-1, intent=-1, val=-1;
      double valf=-1;
      Mshell ashell(argshell, element2);

      ashell.GetContent("id", &id);
      ashell.GetContent("type", &type);
      ashell.GetContent("intent", &intent);
      if (id < 0)
      {
	opid=-1;
	break;
      }

      op->argument[id].type = type;

      if (op->argument[id].intent == SC_INTENT_IN)
      {
	switch (type)
	{
	  case SC_VAR_STREAM:
	    ashell.GetContent("i", &val);
	    trace.GetContent(CStr("/trace/buffers/") << val << "/stream", &val);

	    if (val != op->argument[id].stream)
	    {
	      opid = -1;
	      break;
	    }
	    break;
	  case SC_VAR_INTEGER:
	    ashell.GetContent("d", &val);
	    if (val != op->argument[id].data.d)
	    {
	      opid = -1;
	      break;
	    }
	    break;
	  case SC_VAR_DOUBLE:
	    ashell.GetContent("f", &valf);
	    if (fabs(valf - op->argument[id].data.f) > 0.0001)
	    {
	      opid = -1;
	      break;
	    }
	    break;
	  default:
	    abortErrorl((STREAM(siml.err.GetStream()), "Unknown operation argument type %d", type));
	}
      }

      if (opid == -1)
	break;

      arguments++;
    }
  }

  if (opid == -1)
    abortErrorl((STREAM(siml.err.GetStream()), "Couldn't locate operation signature in trace"));


  /* Mark this operation as "used", so it's not available for subsequent enqueues */
  opshell.ChDir(CStr(opid));
  opshell.MkEntry("used", 1);

  sprintf(buf, "Matched /trace/operations/%d: ", opid);
  opshell.GetContent("text", &buf[strlen(buf)]);
  sprintf(&buf[strlen(buf)], "(");

  /* Write the appropriate output stream identifiers to the stream outputs */
  for (int ii=0; ii != arguments; ++ii)
  {
    if (op->argument[ii].type == SC_VAR_STREAM && op->argument[ii].intent == SC_INTENT_OUT)
    {
      Mshell argshell(opshell, CStr("arguments/") << ii);
      int val;

      argshell.GetContent("i", &val);
      trace.GetContent(CStr("/trace/buffers/") << val << "/stream",
		       &op->argument[ii].stream);
	    
      scuBufferWrite(op->argument[ii].buffer,
		     &op->argument[ii].stream, sizeof(op->argument[ii].stream), 0);
    }

    switch (op->argument[ii].type)
    {
      case SC_VAR_STREAM:
	if (op->argument[ii].intent == SC_INTENT_IN)
	  sprintf(&buf[strlen(buf)], "i [%d], ", op->argument[ii].stream);
	else
	  sprintf(&buf[strlen(buf)], "I [%d], ", op->argument[ii].stream);
	break;
      case SC_VAR_INTEGER:
	if (op->argument[ii].intent == SC_INTENT_IN)
	  sprintf(&buf[strlen(buf)], "i [%d], ", op->argument[ii].data.d);
	else
	  sprintf(&buf[strlen(buf)], "I [%d], ", op->argument[ii].data.d);
	break;
      case SC_VAR_DOUBLE:
	if (op->argument[ii].intent == SC_INTENT_IN)
	  sprintf(&buf[strlen(buf)], "i [%f], ", op->argument[ii].data.f);
	else
	  sprintf(&buf[strlen(buf)], "I [%f], ", op->argument[ii].data.f);
	break;
    }
  }

  buf[strlen(buf)-2] = ')';
  siml.debug() << HDR << buf << std::endl;

  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&scs_operations, op->operation, &so));
  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&so->targets, proc, &st));

  simOperation(proc, op, arguments);
 
  /* Write outputs */
  for (int ii=0; ii != arguments; ++ii)
  {
    if (op->argument[ii].type != SC_VAR_STREAM && op->argument[ii].intent == SC_INTENT_OUT)
    {
      int val;
      double valf;
      Mshell argshell(opshell, CStr("arguments/") << ii);

      switch (op->argument[ii].type)
      {
	case SC_VAR_INTEGER:
	  argshell.GetContent("d", &val);
	  simWriteResult(proc, op, ii, op->argument[ii].type, (char*)&val);
	  break;
	case SC_VAR_DOUBLE:
	  argshell.GetContent("f", &valf);
	  simWriteResult(proc, op, ii, op->argument[ii].type, (char*)&valf);
	  break;
      }      
    }
  }
}

void simOperation(int proc, sim_op_t *op, int arguments)
{
  double bps;
  int bytes, chunksize=0, chunk=0, fri = -1, frs = -1, fbi = -1, fbo = -1;
  long int bytesToSend;
  int masterLength, masterDone=0;
  scs_operation_t *so;
  scs_target_operation_t *st;
  scs_operation_argument_t *sa;
  scs_target_operation_argument_t *sta;
  unsigned char *buf;

  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&scs_operations, op->operation, &so));
  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&so->targets, proc, &st));

  /* Get first of each kind of stream */
  for (int ii=0; ii != arguments; ++ii)
    if (op->argument[ii].type == SC_VAR_STREAM)
    {
      abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, ii, &sta));

      if (sta->timedist == SC_DIST_REGULAR)
      {
	if (frs == -1)
	  frs = ii;
	if (op->argument[ii].intent == SC_INTENT_IN && fri == -1)
	  fri = ii;
      }
      else if (sta->timedist == SC_DIST_BULK)
      {
	if (op->argument[ii].intent == SC_INTENT_IN && fbi == -1)
	  fbi = ii;
	if (op->argument[ii].intent == SC_INTENT_OUT && fbo == -1)
	  fbo = ii;
      }
    }	
  
  /* Get properties of base stream */
  if (fri != -1)
  {
    abortOnNULL(STREAM(siml.err.GetStream()), tblget(&so->arguments, frs, &sa));
    bps = sa->length / st->time;
    masterLength = sa->length;

    abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, frs, &sta));
    chunksize = sta->chunksize;

    siml.debug() << HDR << "Using input " << frs << " (stream "
		 << op->argument[fri].stream << ", " << masterLength << "/" << st->time
		 << " = " << bps << " bps, " << chunksize << " element chunks) as base"
		 << std::endl;
  }
  else if (frs != -1)
  {
    abortOnNULL(STREAM(siml.err.GetStream()), tblget(&so->arguments, frs, &sa));
    bps = sa->length / st->time;
    masterLength = sa->length;
    bytesToSend = masterLength;

    abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, frs, &sta));
    chunksize = sta->chunksize;

    siml.debug() << HDR << "Using output " << frs << " (stream "
		 << op->argument[frs].stream << ", " << masterLength << "/" << st->time
		 << " = " << bps << " bps, " << chunksize << " element chunks) as base"
		 << std::endl;
  }
  else
    bps = 1 / st->time;

  /* Read bulk input streams */
  if (fbi != -1)
  {
    int read, readagain;

    do
    {
      readagain = 0;
      
      for (int ii=0; ii < op->arguments; ii++)
      {
	if (op->argument[ii].type == SC_VAR_STREAM &&
	    op->argument[ii].intent == SC_INTENT_IN)
	{
	  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, ii, &sta));

	  if (sta->timedist == SC_DIST_BULK)
	  {
	    if ((read = scuBufferPeek(op->argument[ii].buffer, op->argument[ii].reader,
				     (void**)&buf,
				     sta->chunksize*sta->elementsize, SCU_BUF_BLOCK_ALL)) ==
		sta->chunksize*sta->elementsize)
	      readagain = 1;

	    if (read > 0)
	      scuBufferReleasePeeked(op->argument[ii].buffer, op->argument[ii].reader,
				     read);
	  }
	}
      }
    } while (readagain);

    /* Clean up bulk input streams */
    for (int ii=0; ii < op->arguments; ii++)
    {
      if (op->argument[ii].type == SC_VAR_STREAM &&
	  op->argument[ii].intent == SC_INTENT_IN)
      {
	abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, ii, &sta));
	
	if (sta->timedist == SC_DIST_BULK)
	  scuBufferRemoveReference(op->argument[ii].buffer);
      }
    }
  }

  for (int ii=0; ii != arguments; ++ii)
    if (op->argument[ii].type == SC_VAR_STREAM)
      op->argument[ii].data.d = 0;

  /* Do fancy regular delay stuff */
  while (1)
  {
    if (fri != -1)
    {
      /* There's an input. Just read from it and delay for the bytes read */
      if (!(bytes = scuBufferPeek(op->argument[fri].buffer,  op->argument[fri].reader,
				  (void**)&buf, chunksize, SCU_BUF_BLOCK_ANY)))
	break;

      scuBufferReleasePeeked(op->argument[fri].buffer,  op->argument[fri].reader, bytes);
    }
    else if (frs != -1)
    {
      /* There's no input; we'll have to keep track of how many bytes we've written. */
      if (!bytesToSend)
	break;

      bytes = MIN(chunksize, bytesToSend);
      bytesToSend -= bytes;
    }
    else
      bytes = 1;

    masterDone += bytes;

    for (int ii=0; ii != arguments; ++ii)
      if (ii != fri && op->argument[ii].type == SC_VAR_STREAM &&
	  op->argument[ii].intent == SC_INTENT_IN)
      {
	abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, ii, &sta));
	if (sta->timedist == SC_DIST_REGULAR)
	{
	  unsigned long long int temp;
	  
	  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&so->arguments, ii, &sa));    
	  
	  temp = (unsigned long long int) sa->length*masterDone/masterLength-op->argument[ii].data.d;
	  op->argument[ii].data.d += temp;
	  
	  scuBufferPeek(op->argument[ii].buffer,  op->argument[ii].reader,
			(void**)&buf, temp, SCU_BUF_BLOCK_ALL);
	  scuBufferReleasePeeked(op->argument[ii].buffer,  op->argument[ii].reader, temp);
	}
      }
    
    unsigned long long int delay = (unsigned long long int)(bytes / bps * 1000000000.0);

    siml.crawl() << HDR << "Processing chunk " << chunk++ << " (" << delay << " us) "
		 << std::endl;

    if (delay < 2 && !isinf(bps))
      siml.warning() << HDR << "Delay is too small for chunk " << chunk-1
		     << ". " << bytes << " / " << bps << " = " << delay << std::endl;

    scos_delay(delay);

    for (int ii=0; ii != arguments; ++ii)
      if (op->argument[ii].type == SC_VAR_STREAM &&
	  op->argument[ii].intent == SC_INTENT_OUT)
      {
	abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, ii, &sta));

	if (sta->timedist == SC_DIST_REGULAR)
	{
	  unsigned long long int temp;
	  
	  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&so->arguments, ii, &sa));    
	  
	  temp = (unsigned long long int) sa->length*masterDone/masterLength-op->argument[ii].data.d;
	  op->argument[ii].data.d += temp;

	  scuBufferAllocate(op->argument[ii].buffer,
			    (void**)&buf, temp, SCU_BUF_BLOCK_ALL);
	  scuBufferReleaseAllocated(op->argument[ii].buffer, temp);
	}
      }
    
    scos_yield();

    if (frs == -1)
      break;
  }

  /* Clean up regular streams */
  for (int ii=0; ii != arguments; ++ii)
  {
    if (op->argument[ii].type == SC_VAR_STREAM)
    {
      abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, ii, &sta));
      if (sta->timedist == SC_DIST_REGULAR)
      {
	if (op->argument[ii].intent == SC_INTENT_IN)
	  scuBufferRemoveReference(op->argument[ii].buffer);
	else
	  scuBufferFinalize(op->argument[ii].buffer);
      }
    }
  }  

  for (int ii=0; ii != arguments; ++ii)
    if (op->argument[ii].type == SC_VAR_STREAM &&
	op->argument[ii].intent == SC_INTENT_OUT)
    {
      abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, ii, &sta));
      if (sta->timedist == SC_DIST_BULK)
	op->argument[ii].data.d = sta->operation_argument->length;
    }

  /* Write bulk output streams */
  if (fbo != -1)
  {
    int write, writeagain;

    do
    {
      writeagain = 0;
      
      for (int ii=0; ii < op->arguments; ii++)
      {
	if (op->argument[ii].type == SC_VAR_STREAM &&
	    op->argument[ii].intent == SC_INTENT_OUT)
	{
	  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, ii, &sta));

	  if (sta->timedist == SC_DIST_BULK)
	  {
	    write = MIN(sta->chunksize*sta->elementsize, op->argument[ii].data.d);

	    if ((write = scuBufferAllocate(op->argument[ii].buffer, (void**)&buf,
				  write, SCU_BUF_BLOCK_ALL)) ==
		sta->chunksize*sta->elementsize)
	      writeagain = 1;

	    if (write > 0)
	      scuBufferReleaseAllocated(op->argument[ii].buffer, write);
	    
	    op->argument[ii].data.d -= write;
	  }
	}
      }
    } while (writeagain);

    /* Clean up bulk output streams */
    for (int ii=0; ii < op->arguments; ii++)
    {
      if (op->argument[ii].type == SC_VAR_STREAM &&
	  op->argument[ii].intent == SC_INTENT_OUT)
      {
	abortOnNULL(STREAM(siml.err.GetStream()), tblget(&st->arguments, ii, &sta));
	
	if (sta->timedist == SC_DIST_BULK)
	  scuBufferFinalize(op->argument[ii].buffer);
      }
    }
  }
}

sim_op_t *simOperationAllocate(void)
{
  int i;

  for (i=0; i < SC_MAX_OPERATIONS; i++)
    if (sim_op[i].id == -1)
    {
      siml.debug() << HDR << "Allocating new operation structure " << i << std::endl;
      sim_op[i].id = i;
      return &sim_op[i];
    }

  siml.err() << HDR << "Not enough room for new operation structure" << std::endl;  

  return NULL;  
}

sim_op_t *simOperationFind(int id)
{
  int i;

  for (i=0; i < SC_MAX_OPERATIONS; i++)
    if (sim_op[i].id == id)
      return &sim_op[i];

  return NULL;
}

void simOperationFree(sim_op_t *op)
{
  op->id = -1;
}

void simWriteResult(int proc, sim_op_t *o, int id, int type, char *data)
{
  scc_msg_t msg;
 
  memset(&msg, 0, sizeof(scc_msg_t));
  msg.header.source = proc;
  msg.header.destination = SCC_MASTER_PROCESSOR;
  msg.header.size = sizeof(scc_msg_t)-sizeof(scc_hdr_t);
  msg.header.type = SCC_MSGTYPE_OPERATIONRESULT;

  msg.opid = o->id;
  msg.optype = o->operation;
  msg.argid = id;
  msg.argtype = type;
  
  switch (msg.argtype)
  {
    case SC_VAR_INTEGER:
      siml.debug() << HDR << "Sending operation " << msg.opid << " VAR_INTEGER argument "
		   << msg.argid << " with value " << *((int*)data) << std::endl;
      memcpy(msg.data, data, sizeof(int));
      break;
    case SC_VAR_DOUBLE:
      siml.debug() << HDR << "Sending operation " << msg.opid << " VAR_DOUBLE argument " 
		   << msg.argid << " with value " << *((double*)data) << std::endl;
      memcpy(msg.data, data, sizeof(double));
      break;
  }

  sccSendMessage((scc_hdr_t*)&msg);
}

void simOperationComplete(int proc, sim_op_t *o)
{
  scc_msg_t msg;
  
  memset(&msg, 0, sizeof(scc_msg_t));
  msg.header.source = proc;
  msg.header.destination = SCC_MASTER_PROCESSOR;
  msg.header.size = sizeof(scc_msg_t)-sizeof(scc_hdr_t);
  msg.header.type = SCC_MSGTYPE_OPERATIONCOMPLETED;

  msg.opid = o->id;
  msg.optype = o->operation;
  
  sccSendMessage((scc_hdr_t*)&msg);
}

extern "C" void simBufferDestructor(scu_buf_t *buf)
{
  scc_msg_t msg;

  memset(&msg, 0, sizeof(scc_msg_t));
  msg.header.source = -1;
  msg.header.destination = SCC_MASTER_PROCESSOR;
  msg.header.size = sizeof(scc_msg_t)-sizeof(scc_hdr_t);
  msg.header.type = SCC_MSGTYPE_BUFFERCOMPLETED;

  msg.argid = buf-sim_buf;
  msg.argtype = SC_VAR_STREAM;
  memcpy(&msg.data[0], &buf->total, sizeof(int));
  memcpy(&msg.data[sizeof(int)], &buf->counter, sizeof(int));

  sccSendMessage((scc_hdr_t*)&msg);
}

extern "C" unsigned long int simBufferQuery(int buffer)
{
  scu_buf_t *buf = &sim_buf[buffer];

  if (!buf->references)
  {
    siml.warning() << HDR << "Premature query for buffer " << buf << std::endl;
    return 0;
  }
  else
    return scuBufferSpaceAvailable(buf);
}

extern "C" void simTransportReceiver(scc_hdr_t *hdr)
{
  scc_datamsg_t *msg = (scc_datamsg_t*)hdr;
  scu_buf_t *buf = &sim_buf[msg->buffer];

  if (!buf->references)
  {
    siml.notice() << HDR << "Premature data for buffer " << msg->buffer
	          << ", holding back" << std::endl; 
  
    while (!buf->references)
    {
      scos_mutex_lock(&sim_buf_mutex);
      scos_cond_wait(&sim_buf_cond, &sim_buf_mutex);
      scos_mutex_unlock(&sim_buf_mutex);
    }
  }

  switch (hdr->type)
  {
    case SCC_MSGTYPE_BUFFERDATA:
      {
        int bytes = hdr->size-(sizeof(scc_datamsg_t)-sizeof(scc_hdr_t));
        scs_target_t *t;
        scs_target_operation_t *o;
        unsigned char *data;

        if (scuBufferSpaceAvailable(buf) < bytes)
  	  siml.crit() << HDR << "Incoming communication of size "
  	 	      << bytes << " for buffer "
		      << msg->buffer << " ( " << scuBufferSpaceAvailable(buf) << " available) would block"
		      << std::endl;


        abortOnNULL(STREAM(siml.err.GetStream()), tblget(&scs_targets, hdr->destination, &t));

        if (tblget(&t->operations, TRANSPORT, &o))
        {
  	  if (o->time > 0)
 	    scos_delay((unsigned long int) (sim_weight[msg->buffer]*bytes*o->time*1000000000L/2));
	  if (o->energy > 0)
	    scos_dissipate(sim_weight[msg->buffer]*bytes*o->energy/2);
        }

        scuBufferAllocate(buf, (void**)&data, bytes, SCU_BUF_BLOCK_ALL);
        memcpy(data, msg->data, 4);
        scuBufferReleaseAllocated(buf, bytes);

        /* For fair scheduling */
        scos_yield();
      } 
      break;
    case SCC_MSGTYPE_BUFFERDRY:
      scuBufferFinalize(buf);
      break;
    default:
      siml.err() << HDR << "Unknown message type " << hdr->type << std::endl;
  }
}

extern "C" void simTransportTransmitter(int src, int dest, int destbuf, int bytes)
{
  scs_target_t *t;
  scs_target_operation_t *o;

  abortOnNULL(STREAM(siml.err.GetStream()), tblget(&scs_targets, src, &t));

  if (tblget(&t->operations, TRANSPORT, &o))
  {
    if (o->time > 0)
      scos_delay(sim_weight[destbuf]*bytes*o->time*1000000000L/2);
    else
      scos_delay(1);

    if (o->energy > 0)
      scos_dissipate(sim_weight[destbuf]*bytes*o->energy/2);
  }
}
